package com.jltx.zeus.observable

import io.reactivex.Observable
import io.reactivex.Observer
import io.reactivex.disposables.Disposable
import io.reactivex.exceptions.CompositeException
import io.reactivex.exceptions.Exceptions
import io.reactivex.plugins.RxJavaPlugins

/**
 * @author jltxseo
 *         Created by junlintianxia on 2020/05/20.
 */
internal class AsyncMultiCallObservable<T>(private val mCall: AsyncMultiCall<T>) :
    Observable<T>() {
    protected override fun subscribeActual(observer: Observer<in T>) {
        val callback = CallCallback(mCall, observer)
        observer.onSubscribe(callback)
        if (!callback.isDisposed) {
            mCall.enqueue(callback)
        }
    }

    private class CallCallback<T> internal constructor(
        private val mCall: AsyncMultiCall<*>,
        private val mObserver: Observer<in T>
    ) :
        Disposable, AsyncMultiCall.Callback<T> {

        @Volatile
        private var mDisposed = false

        @Volatile
        private var mTerminated = false
        override fun dispose() {
            mDisposed = true
            mCall.cancel()
        }

        override fun isDisposed(): Boolean {
            return mDisposed
        }

        override fun onNext(result: T) {
            if (mDisposed || mTerminated) return
            try {
                mObserver.onNext(result)
            } catch (t: Throwable) {
                Exceptions.throwIfFatal(t)
                if (mTerminated) {
                    RxJavaPlugins.onError(t)
                } else if (!mDisposed) {
                    try {
                        mObserver.onError(t)
                    } catch (inner: Throwable) {
                        Exceptions.throwIfFatal(inner)
                        RxJavaPlugins.onError(CompositeException(t, inner))
                    }
                }
            }
        }


        override fun onError(t: Throwable) {
            if (mDisposed || mTerminated) return
            try {
                mTerminated = true
                mObserver.onError(t!!)
            } catch (inner: Throwable) {
                Exceptions.throwIfFatal(inner)
                RxJavaPlugins.onError(CompositeException(t, inner))
            }
        }

        override fun onComplete() {
            if (mDisposed || mTerminated) return
            try {
                mTerminated = true
                mObserver.onComplete()
            } catch (inner: Throwable) {
                Exceptions.throwIfFatal(inner)
                RxJavaPlugins.onError(inner)
            }
        }

    }

}